package cn.doitedu.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/2
 * @Desc: 学大数据，上多易教育
 *   从用户行为日志中，计算：每 5分钟的 pv和 uv
 **/
public class _05_Query_Window_Aggregate_Demo1 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1ms");


        // {"uid":1,"event_id":"page_load","properties":{"url":"/c","ref":"/x"},"action_time":1704719581000}
        tenv.executeSql(
                        " create table action_kfk (                           "+
                        "   uid           int,                                "+
                        " 	event_id      string,                             "+
                        " 	properties    map<string,string>,                 "+
                        " 	action_time   bigint,                             "+

                        " 	uid2 as uid+10,                                   "+  // 表达式字段
                        " 	eid as upper('jacky'),                            "+  // 表达式字段
                        " 	rt as to_timestamp_ltz(action_time,3) ,           "+  // 表达式字段   timestamp *ROWTIME* 类型,代表： event time
                        " 	pt as proctime(),                                 "+  // 表达式字段   timestamp *PROCTIME* 类型,代表: processing time
                        " 	watermark for rt as rt - interval '0' second      "+  // 利用rt字段，声明 watermark
                        " ) with (                                            "+
                        "   'connector' = 'kafka',                            "+
                        "   'topic' = 'action-log',                               "+
                        "   'properties.bootstrap.servers' = 'doitedu:9092',  "+
                        "   'properties.group.id' = 'g002',                   "+
                        "   'scan.startup.mode' = 'latest-offset',            "+
                        "   'value.format' = 'json',                          "+
                        "   'value.fields-include' = 'EXCEPT_KEY'             "+
                        " )                                                   "
        );



        tenv.executeSql(
                       "with tmp as (                                                       \n" +
                               "SELECT                                                      \n" +
                               "    uid,event_id,properties,action_time,rt,pt               \n" +
                               "FROM action_kfk                                             \n" +
                               "WHERE event_id = 'page_load' )                              \n" +
                               "                                                            \n" +
                               "SELECT                                                      \n" +
                               "  window_start,                                             \n" +
                               "  window_end,                                               \n" +
                               "  count(1) as pv,                                           \n" +
                               "  count(distinct uid) as uv                                 \n" +
                               "FROM TABLE(                                                 \n" +
                               "    TUMBLE(TABLE tmp, DESCRIPTOR(rt),INTERVAL '5' MINUTE )  \n" +
                               ")                                                           \n" +
                               "GROUP BY                                                    \n" +
                               "    window_start,                                           \n" +
                               "    window_end                                                "
               ).print();


    }
}
